#include "CQMyClient.h"
#include <thread>
#include "CQBase.h"

MyClient::MyClient(ip_port_c* ipc)
{
	std::thread th(std::bind(&MyClient::write_mq_task, this));
	th.detach();

	ipc_.ip = ipc->ip;
	ipc_.port = ipc->port;
	cbs_ = { on_read,on_connect,on_disconnect };
	client_ = CQNet()->getClient_Tcp(ipc_.port, ipc_.ip, &cbs_, this);
}

MyClient::MyClient(ip_port_c* ipc, io_service& io)
{
	std::thread th(std::bind(&MyClient::write_mq_task, this));
	th.detach();

	ipc_.ip = ipc->ip;
	ipc_.port = ipc->port;
	cbs_ = { on_read,on_connect,on_disconnect };
	client_ = CQNet()->getClient_Tcp(ipc_.port, io, ipc->ip, &cbs_, this);
}

template<typename T>
void clear_queue(std::queue<T>& t)
{
	std::queue<T> empty;
	std::swap(empty, t);
}

MyClient::~MyClient()
{
	close();
	{
		std::lock_guard<std::mutex> lk(writeMU_);
		clear_queue(writeMQ_);
	}
	{
		std::lock_guard<std::mutex> lk(readMU_);
		clear_queue(readMQ_);
	}
}

void MyClient::on_read(const char* msg, size_t len, void* ptr)
{
	MyClient* mc = (MyClient*)ptr;
	if (!mc) { logs(error_, funcx + "this ptr is null"); return; }
	std::string str = string(msg, len);
	logs(debug_, funcx + str);
	{
		std::unique_lock<std::mutex> lk(mc->readMU_);
		mc->readMQ_.push(mc->parse_read_msg(str));
		mc->readCV_.notify_one();
	}
}

void MyClient::on_connect(void* ptr)
{
	MyClient* cli = (MyClient*)ptr;
	cli->set_net_state(true);
	logs(debug_, funcx);
}

void MyClient::on_disconnect(int error_code, void* ptr)
{
	MyClient* cli = (MyClient*)ptr;
	cli->set_net_state(false);
	logs(debug_, funcx);
}

bool MyClient::write(CQ_MSG& mc)
{
	mc.key_from = this->get_key();
	return client_->do_write(build_write_msg(mc));
}

void MyClient::write_mq(CQ_MSG& mc)
{
	std::unique_lock<std::mutex> lk(writeMU_);
	mc.key_from = this->get_key();
	writeMQ_.push(build_write_msg(mc));
	writeCV_.notify_one();
}

CQ_MSG MyClient::read()
{
	std::unique_lock<std::mutex> lk(readMU_);
	readCV_.wait(lk, [this] { return !readMQ_.empty(); });
	CQ_MSG msg = readMQ_.front();
	readMQ_.pop();

	return msg;
}

void MyClient::close()
{
	is_close_.store(true);
	client_->close();
}

void MyClient::stop_io()
{
	client_->stop_io();
}

int MyClient::get_key()
{
	return client_->get_key();
}

bool MyClient::is_connect()
{
	return isConnect_;
}

void MyClient::set_net_state(bool state)
{
	isConnect_ = state;
}

void MyClient::write_mq_task()
{
	while (!is_close_.load())
	{
		if (!isConnect_) { Sleep(50); continue; }
		std::string msg = "";
		{
			std::unique_lock<std::mutex> lk(writeMU_);
			if (is_close_.load())return;
			writeCV_.wait(lk, [this] { return !writeMQ_.empty(); });
			msg = writeMQ_.front();
			writeMQ_.pop();
		}
		client_->do_write(msg);
	}
}
